package com.fuyun.rpc.server.registry;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import com.fuyun.rpc.common.domain.ServiceMetaInfo;
import com.fuyun.rpc.server.config.RegistryConfig;
import io.etcd.jetcd.*;
import io.etcd.jetcd.options.PutOption;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author: zrq1351
 * @date: 2025/2/15
 */
@Slf4j
public class EtcdRegistry implements Registry {

	/**
	 * 根节点
	 */
	private static final String ETCD_ROOT_PATH = "/fuyun-rpc/";
	/**
	 * 本机注册节点key与注册信息ServiceMetaInfo 避免平反序列化
	 */
	private final ConcurrentHashMap<String, ServiceMetaInfo> localRegisterNodeKeyServiceMetaInfoSet =
		new ConcurrentHashMap<>();

	private Client client;
	private KV kvClient;

	/**
	 * 初始化与注册中心的连接
	 *
	 * @param registryConfig 注册中心的配置信息，包含地址、用户名、密码和连接超时时间等
	 */
	@Override
	public void init(RegistryConfig registryConfig) {
		this.client = Client.builder().endpoints(registryConfig.getAddress())
			.user(ByteSequence.from(registryConfig.getUsername(), StandardCharsets.UTF_8))
			.password(ByteSequence.from(registryConfig.getPassword(), StandardCharsets.UTF_8))
			.connectTimeout(Duration.ofMillis(registryConfig.getTimeout())).build();
		this.kvClient = client.getKVClient();
		heartBeat();
	}

	@Override
	public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
		// 创建 Lease 和 KV 客户端
		Lease leaseClient = client.getLeaseClient();

		// 创建一个 30秒的租约
		long leaseId = leaseClient.grant(30).get().getID();

		// 设置要存储的键值对
		String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
		ByteSequence key = ByteSequence.from(registerKey, StandardCharsets.UTF_8);
		ByteSequence value = ByteSequence.from(JSONUtil.toJsonStr(serviceMetaInfo), StandardCharsets.UTF_8);

		// 将键值对与租约关联起来，并设置过期时间
		PutOption putOption = PutOption.builder().withLeaseId(leaseId).build();
		kvClient.put(key, value, putOption).get();

		// 添加节点信息到本地缓存
		localRegisterNodeKeyServiceMetaInfoSet.put(registerKey, serviceMetaInfo);
	}

	@Override
	public void unRegister(ServiceMetaInfo serviceMetaInfo) {
		String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
		kvClient.delete(ByteSequence.from(registerKey, StandardCharsets.UTF_8));

		localRegisterNodeKeyServiceMetaInfoSet.remove(registerKey);
	}

	@Override
	public void destroy() {
		try {
			for (String key : localRegisterNodeKeyServiceMetaInfoSet.keySet()) {
				try {
					kvClient.delete(ByteSequence.from(key, StandardCharsets.UTF_8)).get();
					localRegisterNodeKeyServiceMetaInfoSet.remove(key);
					log.debug("节点：{}下线", key);
				} catch (Exception e) {
					throw new RuntimeException(key + "节点下线失败");
				}
			}
		} catch (RuntimeException e) {
			throw new RuntimeException(e);
		} finally {
			// 释放资源
			if (kvClient != null) {
				kvClient.close();
			}
			if (client != null) {
				client.close();
			}
		}

	}

	@Override
	public void heartBeat() {
		// 10 秒续签一次
		CronUtil.schedule("*/10 * * * * *", (Task)() -> {
			// 遍历本节点所有的 key
			for (String key : localRegisterNodeKeyServiceMetaInfoSet.keySet()) {
				try {
					ServiceMetaInfo serviceMetaInfo = localRegisterNodeKeyServiceMetaInfoSet.get(key);
					List<KeyValue> keyValues =
						kvClient.get(ByteSequence.from(key, StandardCharsets.UTF_8)).get().getKvs();
					if (CollUtil.isEmpty(keyValues)) {
						log.info("节点：{}已被下线，需重启才能生效", key);
						continue;
					}
					register(serviceMetaInfo);
					log.debug("节点：{}续签成功", key);
				} catch (Exception e) {
					log.error("节点：{}续签失败", key, e);
					throw new RuntimeException(key + "续签失败", e);
				}
			}
		});

		// 支持秒级别定时任务
		CronUtil.setMatchSecond(true);
		CronUtil.start();
	}

}